package com.apps.sdses.flink141.cdc;

import com.apps.sdses.flink141.CustomDebeziumDeserializationSchema;
import com.apps.sdses.flink141.DebeziumRecord;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/7/18
 * desc:
 **/
public class Flink_CDCWithCustomerSchema {
    public static void main(String[] args) throws Exception {

        MySqlSource<DebeziumRecord> mySqlSource = MySqlSource.<DebeziumRecord>builder()
                .hostname("localhost")
                .port(3306)
                .databaseList("mydb") // 设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
                .tableList("mydb.orders,mydb.products") // 设置捕获的表
                .username("root")
                .password("123456")
                .startupOptions(StartupOptions.initial())
                .deserializer(new CustomDebeziumDeserializationSchema(true))
//                .deserializer(new JsonDebeziumDeserializationSchema(true))
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 设置 3s 的 checkpoint 间隔
        env.enableCheckpointing(3000);

//        env
//                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
//                // 设置 source 节点的并行度为 4
//                .setParallelism(4)
//                .print().setParallelism(1); // 设置 sink 节点并行度为 1


        //Kafka Sink
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                .setBootstrapServers("192.168.1.2:9092")
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.builder()
                        .setTopic("drity_data_sink")
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                .build();



        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                .map(DebeziumRecord::toString)
                .sinkTo(kafkaSink);

        env.execute("Print MySQL Snapshot + Binlog");


    }
}
